package com.ly.nethub.netty.tcp

import com.ly.nethub.config.CODE_TASK_CANCEL
import com.ly.nethub.config.CODE_TASK_SEND_FAILED
import com.ly.nethub.config.CODE_TASK_TIME_OUT
import com.ly.nethub.config.ConnectStatus
import com.ly.nethub.config.HeartbeatMode
import com.ly.nethub.config.NetHubErrorMsg
import com.ly.nethub.interf.ITaskManager
import com.ly.nethub.task.BaseTask
import com.ly.nethub.utils.LogUtils
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeout
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.PriorityBlockingQueue
import java.util.concurrent.atomic.AtomicBoolean
import kotlin.coroutines.resume
import kotlin.coroutines.suspendCoroutine

/**
 * 任务管理
 */
internal class TcpTaskManager(private val netHub: TcpNetHub) : ITaskManager {
    private val mWaitTaskQueue = PriorityBlockingQueue<BaseTask>()
    private val mStartable = AtomicBoolean(false)
    private val mLooperJobs = ConcurrentHashMap<Long, Pair<Job, BaseTask>>()
    private val mScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
    private val mode get() = netHub.netHubOptions.heartbeatMode

    private val mIsConnected = AtomicBoolean(false)

    override suspend fun tryExecute(task: BaseTask) = coroutineScope {
        if (mWaitTaskQueue.firstOrNull { it.taskId == task.taskId } == null) {
            //由于有重试机，入队列前，尝试重新设置deferred
            task.tryInitialDeferred()
            mWaitTaskQueue.offer(task) //加入队列
        }
        taskLoop()
    }

    private fun taskLoop() {
        if (mStartable.get()) {
            mStartable.set(false)
            LogUtils.d("$TAG 任务队列开始遍历")
            while (true) {
                val task = mWaitTaskQueue.poll()
                if (task == null) {
                    LogUtils.d("$TAG 任务队列已空")
                    break
                }
                executeTask(task)
            }
            mStartable.set(mIsConnected.get())
        }
    }


    //启动一个协程，如果发送成功，需要将对应的协程取消或者重新加入队列
    private fun executeTask(task: BaseTask) {
        val taskJob = mLooperJobs[task.taskId]
        //taskId相同
        val job = taskJob?.first
        if (job?.isActive == true) return
        mLooperJobs[task.taskId] = mScope.launch {
            try {
                withTimeout(task.timeout) {
                    val request = task.buildRequest()
                    val sendResult = suspendCoroutine { sendContinuation ->
                        val channel = netHub.mChannel
                        if (channel == null) {
                            LogUtils.w("$TAG writeAndFlush channel is null,please try reconnect")
                            sendContinuation.resume(false)
                        } else {
                            channel.writeAndFlush(request).addListener {
                                LogUtils.d("$TAG takId:${task.taskId},发送结果:${it.isSuccess},${it.cause()}")
                                sendContinuation.resume(it.isSuccess)
                            }
                        }
                    }
                    if (sendResult) {
                        val response = task.deferred.await()
                        LogUtils.w("$TAG taskId:${task.taskId},hasResponse:${response != null}")
                    } else {
                        task.handleTaskError(
                            CODE_TASK_SEND_FAILED,
                            NetHubErrorMsg[CODE_TASK_SEND_FAILED]
                        )
                    }
                }
            } catch (e: Exception) {
                val isTimeout = e is TimeoutCancellationException
                val code = if (isTimeout) CODE_TASK_TIME_OUT else CODE_TASK_SEND_FAILED
                val msg = NetHubErrorMsg[code]
                task.handleTaskError(code, msg)
            }
        } to task
    }

    private suspend fun BaseTask.handleTaskError(code: Int, msg: String) {
        val maxRetryCount = netHub.netHubOptions.maxRetryCount
        LogUtils.w("$TAG 任务请求发送失败taskId:${taskId},msg:$msg")
        //如果允许重发的话，检查重发次数是否到达最大重发次数
        if (isAllowResend && retryCount < maxRetryCount) {
            //需要重发的任务需要将重发次数加1
            increaseRetryCount()
            //重新加入队列
            tryExecute(this)
        } else {
            callback?.onTaskError(code, msg)
        }
    }

    override fun onConnectStatusChanged(status: ConnectStatus) {
        mIsConnected.set(status is ConnectStatus.Connected)
        mStartable.set(mIsConnected.get())
        taskLoop()
    }

    override fun onMsgReceived(msg: Any?) = synchronized(this) {
        var isTaskFound = false
        val iterator = mLooperJobs.iterator()
        while (iterator.hasNext()) {
            val taskJob = iterator.next()
            val task = taskJob.value.second
            val resp = task.isValidResponse(msg)
            if (resp != null) {
                isTaskFound = true
                LogUtils.d(
                    "$TAG onMsgReceived 找到task:${task.taskId} :${task::class.java.simpleName}"
                )
                task.callback?.onTaskSuccess(resp)
                //设置值，结束挂起等待
                task.deferred.complete(resp)
                if (task.isHeartbeat || mode == HeartbeatMode.Isolate) {
                    //更新心跳响应时间
                    netHub.mHeartbeatHub.updateTime()
                }
                mLooperJobs.remove(task.taskId)
                iterator.remove()
                break
            }
        }
        if (!isTaskFound) {
            //将不属于任务队列中的消息转发
            netHub.onDiscardedMsg(msg)
        }
    }


    override fun destroy() {
        //销毁告知所有任务被取消
        val iterator = mWaitTaskQueue.iterator()
        while (iterator.hasNext()) {
            val task = iterator.next()
            task.callback?.onTaskError(
                CODE_TASK_CANCEL,
                NetHubErrorMsg[CODE_TASK_CANCEL]
            )
            mLooperJobs.remove(task.taskId)
            iterator.remove()
        }
        mStartable.set(false)
    }

    companion object {
        const val TAG = "TaskManager"
    }

}